drop procedure if exists sm_rdd.prc_rdd_work();
create or replace procedure sm_rdd.prc_rdd_work()
as
$$
declare -- here
  v_work_no                 int;
  v_work_job_no             int;
  v_work_lambda_id          varchar(32);
  -- -- v_lambda_name             varchar(64);
  v_lambda_sql              text;
  v_node_id                 varchar(64)    :=   (select cfg_value from sm_rdd.tb_dic_config where cfg_key = 'local_node_id' limit 1);
  v_master_dblink_cfg       text           :=   (select cfg_value from sm_rdd.tb_dic_config where cfg_key = 'master_dblink_cfg' limit 1);
  v_heartbeat_cntdown       int            :=   10;
  v_work_status             varchar(8);
  v_lambda_params           varchar(64)[];
  v_lambda_code             text         ;
  v_lambda_params_valued    text         ;

begin  
  set local lock_timeout = '1s';

  while true
  loop
    v_work_status := null;
    -- 轮询领取 work 
    -- 事务开始; 
    begin
      select 
        work_no          ,
        work_job_no      ,
        work_lambda_id   
      into
        v_work_no      ,
        v_work_job_no  ,
        v_work_lambda_id
      from sm_rdd.tb_trs_work_list 
      where work_no = (select work_no from sm_rdd.tb_trs_work_list where work_status = 'init' limit 1) 
      for update nowait
      ;
      update sm_rdd.tb_trs_work_list
      set work_status = 'doing'
      where work_no = v_work_no
      ;
      commit;
    end;
    -- 事务结束
    if v_work_no is not null
    then
      -- -- -- -- -- 占用节点，采用dblink方式更新，仅锁行，而不是采用fdw表更新，避免更新外表锁表
      -- -- -- -- update sm_rdd.tb_fdw_snap_cluster_node_list
      -- -- -- -- set node_status = '2'
      -- -- -- -- where node_id = v_node_id::int
      -- -- -- -- ;
      perform dblink(v_master_dblink_cfg, 'update sm_rdd.tb_snap_cluster_node_list set node_status = ''2'' where node_id = ' || v_node_id);

      -- 提取 lambda 算子
      select 
        lambda_params  into  v_lambda_params,
        lambda_code    into  v_lambda_code  
      from sm_rdd.tb_fdw_snap_lambda
      where lambda_id = v_work_lambda_id
      ;
      
      begin
        -- 执行 work, 并提交结果
        -- 整理 i_frame 作为入参        
        select 
          string_agg(v_lambda_params[a_idx] || ' := ' || work_input_frame[a_idx] || ';', chr(13) || chr(10)) 
        into 
          v_lambda_params_valued 
        from sm_rdd.tb_sour_job_data, generate_series(1, array_length(v_lambda_params, 1)) tb_a(a_idx)
        where work_no = v_work_no
        ;
        
        v_lambda_sql :=      
          -- lambda 代码，生成匿名代码块，动态语句执行
          '
            do
            $lambda$
              declare 
                ' || v_lambda_params_valued || '
              begin 
                insert into sm_rdd.tb_fdw_sour_work_data
                (
                  work_no,
                  work_job_no_ex,
                  reduce_key,
                  work_output_frame
                )
                with 
                cte_lambda as 
                (
                  ' || lambda_code || '
                )
                select 
                  work_no,
                  ' || v_work_job_no || ',
                  flat_map_key,
                  ''{'' || (regexp_matches(concat(tb_a_ret.*), ''\((.*)\)''))[1] || ''}''
                from sm_rdd.tb_sour_job_data
                  , cte_lambda tb_a_ret(a_ret)
                where work_no = ' || v_work_no || '
              end
            $lambda$
          '
        ;
        
        execute v_lambda_sql;
        
        v_work_status := 'done';
        
        exception when others then
          v_work_status := 'failed';
          -- -- -- logging for error
          raise exception 'error!  %', SQLERRM;
      end;
      
      -- 记录并汇报 work 完成情况
      update sm_rdd.tb_trs_work_list
      set work_status = v_work_status
      where work_no = v_work_no
      ;
      commit;
      -- -- -- -- update sm_rdd.tb_fdw_trs_distrb_job
      -- -- -- -- set job_done_work_cnt = job_done_work_cnt + 1
      -- -- -- -- where job_no = v_work_job_no
      -- -- -- -- ;

      -- -- -- sql 和 while 语句还没能实现 Asynchronous notification 的监听阻塞
      -- -- -- 如果 sm_rdd.ft_rdd_map_output 的 while 轮询做 listen 改造，这里要改造增加步骤，由 listen-notify 的 notify 或 pg_notify 触发 job 完成消息，示例如下：
      -- -- -- -- select dblink_connect('conn', 'hostaddr=192.168.3.105 port=5432 dbname=postgres user=postgres password=seanguo_007')
      -- -- -- -- select dblink_exec('conn', 'listen aaa');
      -- -- -- -- select * from dblink_get_notify('conn');
      -- -- -- -- notify aaa, 'im bbb'
      -- -- -- -- notify aaa, 'im ccc'
      -- -- -- -- notify aaa, 'im ddd'
      -- -- -- -- select * from dblink_get_notify('conn');   -- 阅后即焚
      -- -- -- -- select dblink_exec('conn', 'unlisten aaa');
      -- -- -- -- select dblink_disconnect('conn')

      perform dblink(v_master_dblink_cfg, 'update sm_rdd.tb_trs_distrb_job set job_done_work_cnt = job_done_work_cnt + ' || case when v_work_status = 'done' then '1' else '0' end || ' where job_no = ' || v_work_job_no::varchar(64));

      -- -- -- -- -- 释放节点
      -- -- -- -- update sm_rdd.tb_fdw_snap_cluster_node_list
      -- -- -- -- set node_status = '1'
      -- -- -- -- where node_id = v_node_id::int
      -- -- -- -- ;
      perform dblink(v_master_dblink_cfg, 'update sm_rdd.tb_snap_cluster_node_list set node_status = ''1'', last_free_heartbeat_time = now() where node_id = ' || v_node_id);
    else
      v_heartbeat_cntdown := v_heartbeat_cntdown % 10 + 1;
      if v_heartbeat_cntdown = 10
      then
        perform dblink(v_master_dblink_cfg, 'update sm_rdd.tb_snap_cluster_node_list set node_status = ''1'', last_free_heartbeat_time = now() where node_id = ' || v_node_id);
      end if;
    end if;

    perform pg_sleep(0.2);
  end loop;
end
$$
language plpgsql
;

-- call sm_rdd.prc_rdd_work()

-- # shell 脚本 
-- su - postgres
-- nohup psql -c "call sm_rdd.prc_rdd_work()" > /dev/null &